use crate::keys::VddKey;
use crate::record_db::{load_record, UniKeyTraitExtern};
use crate::{
    cache, delete_by_key_head, load_by_key_head_with_key, move_to_trash, move_to_trash_by_key_head,
    user_db, CF_FV, CF_TRASH, CF_UNIQUE, ROCKSDB,
};
use rust_rocksdb::WriteBatchWithTransaction;
use serde_json::{json, Value};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::u8;
use visit_dd_core::field::VddFieldType;
use visit_dd_core::model::VddSchema;
use visit_dd_core::query::VddValueEvaluation;
use visit_dd_core::record::{RecordRule, RecordRuleDetail, VddDisplay, VddRecord};
use visit_dd_core::value::{VddValue, VddValueWithMeta};
use visit_dd_core::{
    extract_u64_from_key, to_sha256_hash, Ares, Astr, FieldId, RecordId, EMPTY_RID,
};

pub(crate) trait VddValueDdTrait {
    /// 当字段本身定义只允许唯一值的时候，生成字段值的唯一键
    fn get_unique_key(&self, field_id: &FieldId) -> String;
    /// 从列值中形成展示用的字符串
    async fn to_display(&self, field_id: &FieldId, schema: &VddSchema) -> Astr;
}

impl VddValueDdTrait for VddValue {
    fn get_unique_key(&self, field_id: &FieldId) -> String {
        let hash = to_sha256_hash(self.get_simple_display().as_bytes());
        VddKey::UniqueFieldValue(field_id.clone(), hash.into()).to_string()
    }
    async fn to_display(&self, field_id: &FieldId, schema: &VddSchema) -> Astr {
        // 有些系统性质的字段schema中没有
        match schema.try_get_field(field_id) {
            Ok(field) => match field.field_type.as_ref() {
                VddFieldType::FtModel {
                    provider,
                    display_field,
                } => match self {
                    VddValue::S901(v) => {
                        let s_provider = cache::get_schema(*provider).await.unwrap();
                        let provider_record = load_record(&s_provider, v, &[]);
                        if let Some(p_rec) = provider_record {
                            Box::pin(
                                p_rec
                                    .get_value(display_field)
                                    .to_display(display_field, &s_provider),
                            )
                            .await
                        } else {
                            "".into()
                        }
                    }
                    VddValue::M901(v) => {
                        let mut displays = Vec::<Astr>::new();
                        let s_provider = cache::get_schema(*provider).await.unwrap();
                        for link_record_id in v.iter() {
                            load_record(&s_provider, link_record_id, &[]).map(|link_record| {
                                let display = VddDisplay::from(Arc::new(link_record));
                                displays.push(display.get_display(display_field));
                            });
                        }
                        json!(displays).to_string().into()
                    }
                    _ => {
                        panic!("字段类型与值类型不匹配")
                    }
                },
                VddFieldType::FtUserId | VddFieldType::FtUserSelf => match self {
                    VddValue::S911(v) => {
                        let app_user = user_db::u_load(v, false, false).await;
                        app_user.name
                    }
                    _ => {
                        panic!("字段类型与值类型不匹配")
                    }
                },
                VddFieldType::FtCode(code_source) => {
                    let j = serde_json::from_str::<Value>(code_source).unwrap();
                    let codes = j.as_array().unwrap();
                    match self {
                        VddValue::S801(v) => {
                            let mut display = "".to_string();
                            codes.iter().for_each(|x| {
                                if x["c"].as_str().is_some() && x["c"].as_str().unwrap() == v {
                                    display = x["l"].as_str().unwrap().to_string();
                                }
                            });
                            display.into()
                        }
                        VddValue::M801(v) => {
                            let mut displays = Vec::<String>::new();
                            v.iter().for_each(|x| {
                                let mut found = false;
                                codes.iter().for_each(|y| {
                                    if y["c"].as_str().is_some() && y["c"].as_str().unwrap() == x {
                                        displays.push(y["l"].as_str().unwrap().to_string());
                                        found = true;
                                    }
                                });
                            });
                            json!(displays).to_string().into()
                        }
                        _ => {
                            panic!("字段类型与值类型不匹配")
                        }
                    }
                }
                VddFieldType::FtModelSelf(display_field) => match self {
                    VddValue::S903(v) => {
                        let provider_record = load_record(schema, v, &[]);
                        if let Some(p_rec) = provider_record {
                            Box::pin(
                                p_rec
                                    .get_value(display_field)
                                    .to_display(display_field, schema),
                            )
                            .await
                        } else {
                            "".into()
                        }
                    }
                    VddValue::M903(v) => {
                        let mut displays = Vec::<Astr>::new();
                        for link_record_id in v.iter() {
                            load_record(schema, link_record_id, &[]).map(|link_record| {
                                let display = VddDisplay::from(Arc::new(link_record));
                                displays.push(display.get_display(display_field));
                            });
                        }
                        json!(displays).to_string().into()
                    }
                    _ => {
                        panic!("字段类型与值类型不匹配")
                    }
                },
                _ => self.get_simple_display(),
            },
            Err(..) => self.get_simple_display(),
        }
    }
}

/// 检查唯一键，不做任何更改，如果唯一键存在，返回键值, 如果没有发现，返回0
pub(crate) fn unique_check(uk: &str) -> u64 {
    let cf = ROCKSDB.cf_handle(CF_UNIQUE).unwrap();
    let v = ROCKSDB.get_cf(&cf, uk).unwrap();
    match v {
        Some(v_) => u64::from_le_bytes(v_.as_slice().try_into().unwrap()),
        None => 0,
    }
}

/// 加载一个列的一个值
pub fn fv_load(fid: &FieldId, rid: &RecordId) -> Result<VddValue, Astr> {
    let cf_fv = ROCKSDB.cf_handle(CF_FV).unwrap();
    let kh = VddKey::BoundFieldValue(fid.clone(), rid.clone()).to_string();
    let v = ROCKSDB.get_cf(&cf_fv, &kh).unwrap();
    if v.is_none() {
        Err("字段值未找到".into())
    } else {
        let v = serde_json::from_slice::<VddValue>(&v.unwrap()).unwrap();
        Ok(v)
    }
}

/// 从rocksdb中加载列值
pub fn fv_select_all(fid: &FieldId) -> Result<Vec<VddValueWithMeta>, Astr> {
    let kh = VddKey::BoundFieldValue(fid.clone(), EMPTY_RID);
    let mut r = Vec::<VddValueWithMeta>::new();
    load_by_key_head_with_key(&kh).iter().for_each(|fv_tuple| {
        let v = serde_json::from_slice::<VddValue>(&fv_tuple.1).unwrap();
        let id = extract_u64_from_key(&fv_tuple.0);
        r.push(VddValueWithMeta {
            value: Arc::new(v),
            rec_id: RecordId(id),
        });
    });
    Ok(r)
}

/// 从记录中删除一个列的值
pub(crate) async fn fv_delete(
    fid: &FieldId,
    rec_id: &RecordId,
    schema: &VddSchema,
    v_rules: &[Arc<RecordRule>],
) -> Ares {
    let record_old = load_record(schema, &rec_id, &[]).unwrap();
    let k = VddKey::BoundFieldValue(fid.clone(), rec_id.clone());
    let fv = record_old.get_value(fid);
    let cf_u = ROCKSDB.cf_handle(CF_UNIQUE).unwrap();
    let mut batch = WriteBatchWithTransaction::<true>::default();
    if schema.fields.iter().find(|f| &f.id == fid).unwrap().unique {
        // 删除这个值占用的unique key
        let uk = fv.get_unique_key(fid);
        batch.delete_cf(&cf_u, &uk);
    }
    for v_rule in v_rules {
        match v_rule.detail.as_ref() {
            RecordRuleDetail::RrUnique(field_ids) => {
                if field_ids.contains(fid) {
                    let uk = record_old.get_unique_key_by_fids(&v_rule);
                    batch.delete_cf(&cf_u, &uk);
                    // TODO 用字段新null值代替旧的唯一键
                }
            }
            _ => {}
        }
    }
    move_to_trash(&mut batch, &k, None);
    ROCKSDB.write(batch).map_err(|e| e.to_string().into())
}

pub(crate) fn fv_update(
    fid: &FieldId,
    rec_id: &RecordId,
    fv: Arc<VddValue>,
    schema: &VddSchema,
    v_rules: &[Arc<RecordRule>],
) -> Ares {
    let record_old = load_record(schema, &rec_id, &[]).unwrap();
    let k = VddKey::BoundFieldValue(fid.clone(), rec_id.clone()).to_string();
    let fv_old = record_old.get_value(fid).clone();
    let cf_u = ROCKSDB.cf_handle(CF_UNIQUE).unwrap();
    let cf_fv = ROCKSDB.cf_handle(CF_FV).unwrap();
    let mut batch = WriteBatchWithTransaction::<true>::default();
    batch.put_cf(&cf_fv, &k, json!(&fv).to_string());
    if schema.fields.iter().find(|f| &f.id == fid).unwrap().unique {
        // 删除这个值占用的unique key
        let uk = fv_old.get_unique_key(fid);
        batch.delete_cf(&cf_u, &uk);
        let uk = fv.get_unique_key(fid);
        batch.put_cf(&cf_u, &uk, rec_id.0.to_le_bytes());
    }
    for v_rule in v_rules {
        match v_rule.detail.as_ref() {
            RecordRuleDetail::RrUnique(field_ids) => {
                if field_ids.contains(fid) {
                    let uk = record_old.get_unique_key_by_fids(&v_rule).clone();
                    batch.delete_cf(&cf_u, &uk);
                    let mut record_new = VddRecord::new_default(
                        record_old.meta.uid.clone(),
                        record_old.meta.model_id,
                    );
                    record_new.value.insert(fid.clone(), fv.clone());
                    let uk = record_new.get_unique_key_by_fids(&v_rule);
                    batch.put_cf(&cf_u, uk, rec_id.0.to_le_bytes());
                }
            }
            _ => {}
        }
    }
    ROCKSDB.write(batch).map_err(|e| e.to_string().into())
}

/// 删除一个列的所有列值
pub(crate) fn fv_delete_all(batch: &mut WriteBatchWithTransaction<true>, fid: &FieldId) {
    // 删除字段值
    let key_head = VddKey::BoundFieldValue(fid.clone(), EMPTY_RID);
    move_to_trash_by_key_head(batch, &key_head);
    // 删除字段值唯一性约束
    let key_head = VddKey::UniqueFieldValue(fid.clone(), "".into());
    delete_by_key_head(batch, &key_head);
}

/// 将一批记录值从回收站恢复出去 不管模型、字段、记录基础信息
pub(crate) fn fv_restore(field_ids: &[FieldId], sub_record_ids: Option<&[RecordId]>) -> Ares {
    let cf_trash = ROCKSDB.cf_handle(CF_TRASH).unwrap();
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut iter = ROCKSDB.raw_iterator_cf(&cf_trash);
    iter.seek_to_first();
    let cf_fv = ROCKSDB.cf_handle(CF_FV).unwrap();
    while iter.valid() {
        let (k, v) = iter.item().unwrap();
        let key = String::from_utf8(k.to_vec()).unwrap();
        let key = VddKey::try_from(key.as_str());
        if let Ok(key) = key {
            let key_s = key.to_string();
            match key {
                VddKey::BoundFieldValue(f, r) => {
                    if field_ids.contains(&f) {
                        if sub_record_ids.is_none() || sub_record_ids.unwrap().contains(&r) {
                            // TODO 这里放回去的时候，唯一性约束还没校验，应该先校验唯一性约束再操作
                            // let vdd_value = serde_json::from_slice::<VddValue>(v).unwrap();
                            batch.put_cf(&cf_fv, &key_s, v);
                            batch.delete_cf(&cf_trash, &key_s);
                            // 列值往回收站放的时候没有动列值缓存，所以恢复的时候也不需要动列值缓存
                            // cache::on_fv_added(&fid, &rec_id, &Arc::new(vdd_value));
                        }
                    }
                }
                _ => {}
            }
        }
        iter.next();
    }
    ROCKSDB.write(batch).map_err(|e| e.to_string().into())
}

/// 唯一值保证的改变情况
#[derive(Debug)]
pub struct UniqueChange {
    pub a: u8,
    pub key: Astr,
    pub value: u64,
}

/// 应用一批唯一值的改变
pub(crate) fn apply_unique_change(changes: &[UniqueChange]) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    changes.iter().for_each(|change| {
        tracing::warn!("{:?}", change);
        let cf_u = ROCKSDB.cf_handle(CF_UNIQUE).unwrap();
        if change.a == u8::MAX {
            batch.delete_cf(&cf_u, change.key.as_bytes());
        } else {
            batch.put_cf(&cf_u, change.key.as_ref(), change.value.to_le_bytes());
        }
    });
    ROCKSDB.write(batch).map_err(|e| e.to_string().into())
}

/// 从列值中找到符合条件的记录id列表
pub(crate) fn select_rec_ids_by_query(p: &[Arc<VddValueEvaluation>]) -> HashSet<RecordId> {
    let mut id_map = HashMap::<RecordId, usize>::new();
    p.iter().for_each(|vvef| {
        let field_id = vvef.get_field_id();
        fv_select_all(&field_id).unwrap().iter().for_each(|fvm| {
            if vvef.evaluate(&fvm.value) {
                let mut hit_count = *id_map.get(&fvm.rec_id).unwrap_or(&0usize);
                hit_count += 1;
                id_map.insert(fvm.rec_id.clone(), hit_count);
            }
        })
    });
    // tracing::warn!("{:?}", id_map);
    id_map
        .iter()
        .filter(|kv| *kv.1 == p.len())
        .map(|kv| kv.0.clone())
        .collect()
}
